﻿using System;
using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using System.IO;
using System.Net.Sockets;
using Polly;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;

namespace Fast.Extensions.RabbitMQ
{
    /// <summary>
    /// 持久化连接对象
    /// </summary>
    public class PersistentConnection : IPersistentConnection
    {
        private readonly ConnectionFactory _connectionFactory;
        private readonly ILogger<PersistentConnection> _logger;
        private readonly int _retryCount;
        private IConnection _connection;
        private bool _isDisposed;

        private readonly object _syncRoot = new object();

        public PersistentConnection(ConnectionFactory connectionFactory, ILoggerFactory loggerFactory,
            int retryCount = 5)
        {
            _connectionFactory = connectionFactory ?? throw new ArgumentNullException(nameof(connectionFactory));
            _logger = (loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory)))
                .CreateLogger<PersistentConnection>();
            _retryCount = retryCount;
        }

        /// <summary>
        /// 是否已连接
        /// </summary>
        public bool IsConnected => _connection is {IsOpen: true} && !_isDisposed;

        /// <summary>
        /// 连接
        /// </summary>
        /// <returns></returns>
        public bool TryConnect()
        {
            _logger.LogInformation("RabbitMQ 客户端正在尝试连接...");

            lock (_syncRoot)
            {
                var policy = Policy.Handle<SocketException>().Or<BrokerUnreachableException>()
                    .WaitAndRetry(_retryCount, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)),
                        (ex, time) =>
                        {
                            _logger.LogWarning(ex, "RabbitMQ 客户端在 {TimeOut} 秒之后无法连接 ({ExceptionMessage})",
                                $"{time.TotalSeconds:n1}", ex.Message);
                        }
                    );
                var hostnames = _connectionFactory.HostName.TrimEnd(';').Split(';');
                policy.Execute(() =>
                {
                    _connection = hostnames.Length == 1
                        ? _connectionFactory.CreateConnection()
                        : _connectionFactory.CreateConnection(hostnames);
                });

                if (IsConnected)
                {
                    _connection.ConnectionShutdown += OnConnectionShutdown;
                    _connection.CallbackException += OnCallbackException;
                    _connection.ConnectionBlocked += OnConnectionBlocked;

                    _logger.LogInformation("RabbitMQ 客户端获得了 '{HostName}' 的持久连接，并订阅了失败事件",
                        _connection.Endpoint.HostName);

                    return true;
                }
                else
                {
                    _logger.LogCritical("致命错误：无法创建和打开RabbitMQ连接");

                    return false;
                }
            }
        }

        /// <summary>
        /// 创建信道
        /// </summary>
        /// <returns></returns>
        public IModel CreateModel()
        {
            if (!IsConnected)
            {
                throw new InvalidOperationException("没有可用于执行此操作的RabbitMQ连接");
            }

            return _connection.CreateModel();
        }

        /// <summary>
        /// 释放连接
        /// </summary>
        public void Dispose()
        {
            if (_isDisposed) return;

            _isDisposed = true;

            try
            {
                _connection.Dispose();
            }
            catch (IOException ex)
            {
                _logger.LogCritical(ex.ToString());
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex.ToString());
            }
        }

        #region Private Methods

        /// <summary>
        /// RabbitMQ连接Shutdown处理
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="reason"></param>
        private void OnConnectionShutdown(object sender, ShutdownEventArgs reason)
        {
            if (_isDisposed) return;

            _logger.LogWarning("RabbitMQ连接正在关闭。正在尝试重新连接...");

            TryConnect();
        }

        /// <summary>
        /// RabbitMQ连接回调时发生异常处理
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnCallbackException(object sender, CallbackExceptionEventArgs e)
        {
            if (_isDisposed) return;

            _logger.LogWarning("RabbitMQ连接引发异常。正在尝试重新连接...");

            TryConnect();
        }

        /// <summary>
        /// RabbitMQ连接被阻断时处理
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnConnectionBlocked(object sender, ConnectionBlockedEventArgs e)
        {
            if (_isDisposed) return;

            _logger.LogWarning("RabbitMQ连接已关闭。正在尝试重新连接...");

            TryConnect();
        }

        #endregion
    }
}